/*
 * compare pthread_cond_t and select/read/write notify speed.
 * and also compare pipe, socketpair, eventfd
 *
 * PS: http://bbs.chinaunix.net/thread-280922-1-1.html
 */

#include <unistd.h>
#include <sys/eventfd.h>
#include <errno.h>
#include <sys/time.h>
#include <sys/types.h>
#include <sys/select.h>
#include <stdlib.h>
#include <stdio.h>
#include <stdint.h>
#include <thread>
#include <iostream>
#include <functional>
#include <string>
#include <sys/types.h>
#include <sys/socket.h>
#include <sched.h>
#include <sys/syscall.h>


#define PVAR(x) do { std::cout << #x ": " << (x)  << std::endl; } while(false)
#define PERR(fmt, ...) do { fprintf( stderr, "ERROR:" #fmt "\n", ##__VA_ARGS__); exit(EXIT_FAILURE); } while(false)

class BaseCond {
public:
	virtual void wait(void) = 0;
	virtual void notify(void) = 0; // notify one
};

class ThreadCond : public BaseCond{
public:
	ThreadCond() {
		flag_ = false;
	}
	virtual void notify(void) {
		std::lock_guard<std::mutex> lm(m_);
		flag_ = true;
		c_.notify_one();
	}

	virtual void wait(void) {
		std::unique_lock<std::mutex> lm(m_);
		while(!flag_) {
			c_.wait(lm);
		}
		flag_ = false;
	}

private:
	std::mutex m_;
	std::condition_variable c_;
	volatile bool flag_;
};

class FileCond : public BaseCond {
public:
	FileCond() : fd_{0, 0}, read_buf_(0), write_buf_(1), buf_size_(sizeof(uint64_t)) {}
	virtual ~FileCond()
	{
		if(fd_[0] > 0)
			close(fd_[0]);
		if(fd_[1] > 0 && fd_[1] != fd_[0])
			close(fd_[1]);
	}

	virtual void notify(void)
	{
		int rc = write(fd_[1], &write_buf_, buf_size_);
		if(buf_size_ != rc) {
			PERR("write error, rc %d, errno %d", rc, errno);
		}
	}
	virtual void wait(void)
	{
		FD_ZERO(&fds_);
		FD_SET(fd_[0], &fds_);
		int rc = select(fd_[0] + 1, &fds_, NULL, NULL, NULL);
		if( rc < 0 ) {
			PERR( "select failed, errno %d", errno);
		}
		if(FD_ISSET(fd_[0], &fds_)) {
			int rc = read(fd_[0], &read_buf_, buf_size_);
			if( rc < buf_size_ ) {
				PERR( "read failed, rc %d, errno %d", rc, errno );
			}
		}
		else {
			PERR( "this should not happen." );
		}
	}

protected:
	int fd_[2]; // fd_[0] for read, fd_[1] for write

	uint64_t read_buf_;
	uint64_t write_buf_;

	int buf_size_;

private:
	fd_set fds_;
};

class EventfdCond : public FileCond {
public:
	EventfdCond() {
		int fd = eventfd(0, 0);
		if( fd < 0 ) {
			PERR( "eventefd failed, fd %d, errno %d", fd, errno );
		}
		fd_[0] = fd;
		fd_[1] = fd;
	}
	virtual ~EventfdCond() {}
};

class PipeCond : public FileCond {
public:
	PipeCond()
	{
		if( pipe(fd_) < 0 ) {
			PERR( "pipe failed, errno %d", errno );
		}

		buf_size_ = 1;
	}
	virtual ~PipeCond() {}
};

class SocketpairCond : public FileCond {
public:
	SocketpairCond()
	{
		if( socketpair(AF_UNIX, SOCK_STREAM, 0, fd_) < 0 ) {
			PERR( "socketpair failed, errno %d", errno );
		}

		buf_size_ = 1;
	}
	virtual ~SocketpairCond() {}
};

class CondFactory {
public:
	static BaseCond *create(const std::string &name) {
		if( name == "thread" )
			return new ThreadCond();
		else if(name == "eventfd" )
			return new EventfdCond();
		else if(name == "pipe" )
			return new PipeCond();
		else if(name == "socketpair" )
			return new SocketpairCond();
		else {
			PERR( "unknown name %s", name.c_str());
			return NULL;
		}
	}
};

void master_thread(BaseCond *m2s_cond, BaseCond *s2m_cond, cpu_set_t *cs, uint64_t times)
{
	if( sched_setaffinity( (pid_t)syscall(SYS_gettid), sizeof(*cs), cs) < 0 )
		PERR( "sched_setaffinity failed, errno %d", errno );

	while(times > 0 ) {
		m2s_cond->notify();
		s2m_cond->wait();
		--times;
		if( times % 200000 == 0 && times > 0 )
			PVAR(times);
	}
	exit(EXIT_SUCCESS);
}

void slave_thread(BaseCond *m2s_cond, BaseCond *s2m_cond, cpu_set_t *cs)
{
	if( sched_setaffinity( (pid_t)syscall(SYS_gettid), sizeof(*cs), cs) < 0 )
		PERR( "sched_setaffinity failed, errno %d", errno );

	while(true) {
		m2s_cond->wait();
		s2m_cond->notify();
	}
}

int main(int argc, char *argv[])
{
	std::string name = argv[1];

	BaseCond *ms = CondFactory::create(name);
	BaseCond *sm = CondFactory::create(name);

	uint64_t times = atoll(argv[2]);

	cpu_set_t master_cs;
	cpu_set_t slave_cs;
	CPU_ZERO(&master_cs);
	CPU_ZERO(&slave_cs);

	// hard coded
	CPU_SET(atoi(argv[3]), &master_cs);
	CPU_SET(atoi(argv[4]), &slave_cs);

	// std::thread t([=]{master_thread(ms, sm, times);});
	// std::thread t(std::bind(master_thread, ms, sm, times));
	std::thread mt(master_thread, ms, sm, &master_cs, times);
	std::thread st(slave_thread, ms, sm, &slave_cs );

	mt.join();
	st.join();

	// never reach here, and memory leak ... (ms, sm)
}

